package com.xiehua.task.actor;

import akka.actor.AbstractActor;
import akka.actor.ActorSelection;
import com.xiehua.task.actor.cmd.Cmd;
import com.xiehua.task.job.bucket.ReadyBucket;
import com.xiehua.task.job.pool.Job;
import com.xiehua.task.redis.RedisTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class TaskReadyActor extends AbstractActor {

    private static final Logger LOGGER = LoggerFactory.getLogger(TaskReadyActor.class);

    @Override
    public Receive createReceive() {
        return receiveBuilder().match(Cmd.class, t -> t.getCmd().equals(Cmd.CMD_TASK_READY), s -> {
            if(LOGGER.isDebugEnabled()) LOGGER.debug("收到消息type：{}",s.getCmd());
            Object data = s.getT();
            if(data != null){
                List<Job> jobs = (List<Job>) data;
                LOGGER.info("收到待处理的任务为:{}",jobs.size());
                ReadyBucket readyBucket = new ReadyBucket(RedisTemplate.getJedis());
                readyBucket.offerTask(jobs);//批量添加任务
                readyBucket.clear();
                //发布事件
               // getContext().system().eventStream().publish(new Cmd(Cmd.EVT_TASK_DISPOSE,jobs));
                ActorSelection taskNotifyActor = getContext().actorSelection("/user/TaskDisposeActor");
                taskNotifyActor.tell(new Cmd(Cmd.EVT_TASK_DISPOSE,null),self());
            }
        }).matchAny(t->{
            LOGGER.error("没有对应Actor处理本次消息~");
            unhandled(t);
        }).build();
    }
}
